package com.ververica.cdc.guass.source.hybrid;

import com.ververica.cdc.guass.source.jdbc.SnapShotSplit;
import org.apache.flink.api.connector.source.SourceSplit;

import java.util.ArrayList;
import java.util.List;

/**
 * HybridSourceSplit 实现了 SourceSplit 接口，用于在不同源之间进行切换时封装 split 信息。
 */
public class HybridSourceSplit implements SourceSplit {
    private final String splitId;
    private final boolean isSnapshot;
    private final Object splitDetail;
    private final int sourceIndex; // 保留 sourceIndex 字段

    public HybridSourceSplit(String splitId, boolean isSnapshot, Object splitDetail, int sourceIndex) {
        this.splitId = splitId;
        this.isSnapshot = isSnapshot;
        this.splitDetail = splitDetail;
        this.sourceIndex = sourceIndex;
    }

    public boolean isSnapshot() {
        return isSnapshot;
    }

    public Object getSplitDetail() {
        return splitDetail;
    }

    public int getSourceIndex() {
        return sourceIndex;
    }

    @Override
    public String splitId() {
        return splitId;
    }

    // 包装单个 split
    public static HybridSourceSplit wrapSplit(SourceSplit split, int sourceIndex) {
        return new HybridSourceSplit(split.splitId(), split instanceof SnapShotSplit, split, sourceIndex);
    }

    // 包装多个 splits
    public static List<HybridSourceSplit> wrapSplits(List<? extends SourceSplit> splits, int sourceIndex) {
        List<HybridSourceSplit> wrappedSplits = new ArrayList<>();
        for (SourceSplit split : splits) {
            wrappedSplits.add(wrapSplit(split, sourceIndex));
        }
        return wrappedSplits;
    }

    // 解包多个 splits
    public static List<SourceSplit> unwrapSplits(List<HybridSourceSplit> hybridSplits) {
        List<SourceSplit> splits = new ArrayList<>();
        for (HybridSourceSplit hybridSplit : hybridSplits) {
            splits.add((SourceSplit) hybridSplit.getSplitDetail());
        }
        return splits;
    }
}
